package org.qiyu.live.user.provider.service.impl;


import com.alibaba.fastjson.JSON;

import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;
import org.qiyu.live.user.interfaces.dto.UserDTO;
import org.qiyu.live.user.provider.dao.mapper.IUserMapper;
import org.qiyu.live.user.provider.dao.po.UserPO;
import org.qiyu.live.user.provider.service.IUserService;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class UserServiceImpl implements IUserService {


    @Resource
    private IUserMapper userMapper;

    @Resource
    private RedisTemplate<String,UserDTO> redisTemplateSoftware;

    @Resource
    private MQProducer rocketMQProducer;


    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;



    @Override
    public UserDTO getByUserId(Long userId) {
        //1.1 校验参数
        if(userId == null){
            return null;
        }
        //1.2 定义redis的key
        //qiyu-live-user-provider:userInfo:
        String userKey = userProviderCacheKeyBuilder.buildUserInfoKey(userId);


        //1.3 从redis当中获取数据
        UserDTO userDTO = redisTemplateSoftware.opsForValue().get(userKey);
        if(userDTO != null){
            return userDTO;
        }
        //1.4 redis当中没有数据就到MySQL去查用户数据
        UserPO userPO = userMapper.selectById(userId);
        userDTO = ConvertBeanUtils.convert(userPO, UserDTO.class);

        //1.5 将mysql中的用户数据存入到redis中
        if(userDTO != null){
            redisTemplateSoftware.opsForValue().set(userKey,userDTO,30, TimeUnit.MINUTES);
        }

        return userDTO;
    }

    @Override
    public boolean insertOne(UserDTO userDTO) {
        if(userDTO == null || userDTO.getUserId() == null){
            return false;
        }
        UserPO userPO = ConvertBeanUtils.convert(userDTO, UserPO.class);
        userMapper.insert(userPO);
        return true;

    }

    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        if(userDTO.getUserId() == null){
            return false;
        }
        UserPO userPO = ConvertBeanUtils.convert(userDTO, UserPO.class);
        userMapper.updateById(userPO);

        //删除缓存
        String key = userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
        redisTemplateSoftware.delete(key);

        //发送RocketMQ延迟消息
        try{
            Message message = new Message();
            message.setBody(JSON.toJSONString(userDTO).getBytes());
            message.setTopic("user-update-cache");
            message.setDelayTimeLevel(1);
            rocketMQProducer.send(message);
        }catch (Exception e) {
            throw new RuntimeException(e);
        }

        return true;
    }


    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIdList) {
        //1、校验参数
        if(CollectionUtils.isEmpty(userIdList)){
            return new HashMap<>();
        }
        //1、1 校验参数过滤出用户id大于10000的用户
        userIdList = userIdList.stream().filter(x -> x > 10000).collect(Collectors.toList());


        // 构建redis的keyList
        ArrayList<String> userCacheKeyList = new ArrayList<>();
        for (Long userId : userIdList) {
            userCacheKeyList.add(userProviderCacheKeyBuilder.buildUserInfoKey(userId));
        }

        //通过mutiget批量查询用户数据
        List<UserDTO> userDTOList = redisTemplateSoftware.opsForValue().multiGet(userCacheKeyList).stream().filter(x->x!=null).collect(Collectors.toList());
        
        //所有用户数据都已经在redis缓存中查询到，所有直接return
        if(!CollectionUtils.isEmpty(userDTOList)&&userDTOList.size()==userIdList.size()){
            //直接返回给前端
            return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId,x->x));
        }

        //命中缓存的所有用户的id集合
        List<Long> userIdInCacheList  = userDTOList.stream().map(UserDTO::getUserId).collect(Collectors.toList());
        //未命中缓存的所以用户的id集合
        List<Long> userIdNotInCacheList  = userIdList.stream().filter(x -> !userIdInCacheList.contains(x)).collect(Collectors.toList());

        //2、通过多线程+本地内存的方法提高批量查询用户信息的性能
        //2.1 将userIdList与100取模转为map
        Map<Long, List<Long>> userMap = userIdNotInCacheList.stream().collect(Collectors.groupingBy(userId -> userId % 100));

        //2、2 通过并行流parallelStream多线程去mysql里面批量查询用户数据
        List<UserDTO> dbQueryResult  = new CopyOnWriteArrayList<>();
        //2.3 循环到mysql里面去查询用户数据
        userMap.values().parallelStream().forEach(queryUserIdList->{
            dbQueryResult.addAll(ConvertBeanUtils.convertList(userMapper.selectBatchIds(queryUserIdList), UserDTO.class));
        });
        if(!CollectionUtils.isEmpty(dbQueryResult)){

            //map<key,value>  ------>map<qiyu-live-user-provider:userInfo:100001,UserDTO>
            Map<String, UserDTO> saveCacheMap  = dbQueryResult.stream().collect(Collectors.toMap(userDTO -> userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId()), x -> x));
            redisTemplateSoftware.opsForValue().multiSet(saveCacheMap );
            redisTemplateSoftware.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for (String redisKey : saveCacheMap.keySet()) {
                        operations.expire((K) redisKey, createRandomTime(), TimeUnit.SECONDS);
                    }
                    return null;
                }
            });

            userDTOList.addAll(dbQueryResult);

        }

        return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId,x->x));
    }


    /**
     * 创建随机时间
     * @return
     */
    private int createRandomTime(){
        return ThreadLocalRandom.current().nextInt(1000)+30*60;
    }




}
